package cgl.hadoopsensorgrid.sensorgrid.eval;

import java.lang.InterruptedException;
import java.io.IOException;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.BufferedReader;
import java.util.Hashtable;

import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.Text;

import cgl.hadoopsensorgrid.sensorgrid.util.*;

/***
 * 
 * The evalMapper extends the hadoop Mapper class. It is deployed by the
 * ryoController. It gets data from temporary files and passes it to Reducer.
 * 
 * @author Chao Sun
 * 
 */
public class evalMapper extends
		Mapper<Object, Text, serializedText, BytesWritable> {
	private int PIDBase = 1;
	private int IDBase = 1;
	private Hashtable sitesTable;

	/**
	 * Override the setup method. Initialize sitesTable.
	 * 
	 * @param context
	 *          context of this evalMapper object
	 */
	protected void setup(Context context) {
		sitesTable = new Hashtable<String, Integer>();
		System.out.println("Eval mapper started");
	}

	/**
	 * This method overrides the map() method of class Mapper. It gets data from
	 * temporary files and passes it to Reducer.
	 * 
	 * @param key
	 *          key of the data split
	 * @param value
	 *          contents of the data split, which is the filename
	 * @param context
	 *          context of this evalMapper object
	 * 
	 * @throws InterruptedException
	 * 
	 * @throws IOException
	 */
	protected void map(Object key, Text value, Context context)
			throws IOException, InterruptedException {
		String fileName = value.toString();
		if (fileName == null || fileName.length() == 0) {
			System.out.println("Evaluation data temp file path error");
			System.exit(0);
		}
		System.out.println("fileName: " + fileName);
		
		int minLength = 0;
		int offSet = 0;
		byte[] typeBytes = "type".getBytes();
		if (fileName.endsWith("ascii.txt")) {
			minLength = 30;
			offSet = 26;
			typeBytes = "ASCII".getBytes();
		} else if (fileName.endsWith("pos.txt")) {
			minLength = 4;
			offSet = 0;
			typeBytes = "POS".getBytes();
		} else {
			System.out.println("Data type error. Exit");
			System.exit(0);
		}

		BufferedReader dataReader = new BufferedReader(new FileReader(fileName));
		String dataLine;
		while ((dataLine = dataReader.readLine()) != null) {
			if (!dataLine.startsWith("EOF")) {
				if (dataLine.length() > minLength) {
					int PID;
					String siteName = dataLine.substring(offSet, minLength);
					if (sitesTable.containsKey(siteName)) {
						PID = ((Integer) sitesTable.get(siteName)).intValue();
					} else {
						PID = PIDBase++;
						sitesTable.put(siteName, new Integer(PID));
					}

					byte[] siteBytes = siteName.getBytes();
					int ID = IDBase++;
					serializedText keyToReducer = new serializedText();
					keyToReducer.set(PID, siteBytes, ID, typeBytes);
					byte[] dataBytes = dataLine.getBytes();
					BytesWritable valueToReducer = new BytesWritable();
					valueToReducer.set(dataBytes, 0, dataBytes.length);
					context.write(keyToReducer, valueToReducer);
				}
			}
		}
	}

}